package pers.amos.eshop.storm.bolt;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @author amos wong
 * @create 2020-11-23 5:28 下午
 * <p>
 * 日志解析的bolt
 */

@Slf4j
public class LogParseBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        String message = tuple.getStringByField("message");

        log.info("【LogParseBolt接收到一条日志】message={}", message);

        JSONObject messageObject = JSONObject.parseObject(message);
        JSONObject uriArgs = messageObject.getJSONObject("uri_args");
        Long productId = uriArgs.getLong("productId");

        if (productId != null) {
            collector.emit(new Values(productId));
            log.info("【LogParseBolt发射出去一个商品id】productId={}", productId);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("productId"));
    }
}
